package util;

import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.transport.client.PreBuiltTransportClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Map;
import java.util.Properties;

import static org.elasticsearch.index.query.QueryBuilders.termQuery;

/**
 * 使用java API操作elasticSearch
 *
 * @author 231
 *
 */
public class ElasticsearchUtil {

    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchUtil.class);

    private static final Logger ES_ERROR = LoggerFactory.getLogger("esErrorLogger");

    private static final String ES_IP = PropertyUtil.getValue("es.ip");

    private static final int ES_PORT = Integer.valueOf(PropertyUtil.getValue("es.port"));

    private static final long ES_SCROLL_TIMEOUT = Long.valueOf(PropertyUtil.getValue("es.scroll.timeout"));

    /**
     * 获取连接
     * @throws Exception
     */
    public static TransportClient getClient() throws Exception {
        return new PreBuiltTransportClient(Settings.EMPTY)
                .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ES_IP), ES_PORT));
    }

    /**
     * 删除查询到的
     * @param index
     * @param type
     * @param name
     * @param value
     * @throws UnknownHostException
     */
    public static void scrollSearchDelete(String index, String type, String name,String value) throws Exception {
        Client client = getClient();
        QueryBuilder qb = termQuery(name, value);
        SearchResponse scrollResp = client.prepareSearch(index).setTypes(type)
                .setSearchType(SearchType.DEFAULT)
                .setScroll(new TimeValue(ES_SCROLL_TIMEOUT))
                .setQuery(qb)
                .setSize(100).execute().actionGet(); //100 hits per shard will be returned for each scroll

        BulkRequestBuilder bulkRequest = client.prepareBulk();

        while (true) {
            for (SearchHit hit : scrollResp.getHits().getHits()) {
                bulkRequest.add(client.prepareDelete(hit.getIndex(),hit.getType(),hit.getId()));
            }
            scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(ES_SCROLL_TIMEOUT)).execute().actionGet();
            if (scrollResp.getHits().getHits().length == 0) {
                break;
            }
        }
        if (bulkRequest.numberOfActions()>0) {
            BulkResponse bulkResponse = bulkRequest.execute().actionGet();
            if (bulkResponse.hasFailures()) {
                BulkItemResponse[] bulkItemResponse = bulkResponse.getItems();
                for (int i = 0; i < bulkItemResponse.length; i++) {
                    System.out.println(bulkItemResponse[i].getItemId() + ":" + bulkItemResponse[i].getIndex() + ":" + bulkItemResponse[i].getFailureMessage());
                }
            }
        }
        client.close();
    }

    /**
     * 更新查询到的
     * @param index
     * @param name
     * @param value
     * @throws Exception
     */
    public static void scrollSearchUpdate(String index, String type, String name,String value,Map source) throws Exception {
        Client client = getClient();
        QueryBuilder qb = termQuery(name, value);
        SearchResponse scrollResp = client.prepareSearch(index).setTypes(type)
                .setSearchType(SearchType.DEFAULT)
                .setScroll(new TimeValue(ES_SCROLL_TIMEOUT))
                .setQuery(qb)
                .setSize(100).execute().actionGet();

        BulkRequestBuilder bulkRequest = client.prepareBulk();

        while (true) {
            for (SearchHit hit : scrollResp.getHits().getHits()) {
                bulkRequest.add(client.prepareUpdate(hit.getIndex(),hit.getType(),hit.getId())
                .setDoc(source));
            }
            scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(ES_SCROLL_TIMEOUT)).execute().actionGet();
            if (scrollResp.getHits().getHits().length == 0) {
                break;
            }
        }
        if (bulkRequest.numberOfActions()>0){
            BulkResponse bulkResponse = bulkRequest.execute().actionGet();
            if (bulkResponse.hasFailures()) {
                BulkItemResponse[] bulkItemResponse = bulkResponse.getItems();
                for (int i = 0; i <bulkItemResponse.length ; i++) {
                    System.out.println(bulkItemResponse[i].getItemId()+":"+bulkItemResponse[i].getIndex()+":"+bulkItemResponse[i].getFailureMessage());
                }
            }
        }
        client.close();
    }



}